package com.upcoming.data.storm.util;


import org.apache.kafka.common.serialization.Deserializer;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.Map;

/**
 * 作者：tianjiayuan
 * 创建时间：2018-05-13 18:54
 * 类描述：
 * 修改人：
 * 修改时间：
 */
public class EventDeserializer implements Deserializer<Event> {

	@Override
	public void configure(Map<String, ?> map, boolean b) {

	}

	@Override
	public Event deserialize(String s, byte[] bytes) {
		return BytesToObject(bytes);
	}

	@Override
	public void close() {

	}


	/**
	 * 字节数组转对象
	 *
	 * @param bytes
	 * @return
	 */
	private static Event BytesToObject(byte[] bytes) {
		Event event = null;
		ByteArrayInputStream bi = null;
		ObjectInputStream oi = null;
		try {
			bi = new ByteArrayInputStream(bytes);
			oi = new ObjectInputStream(bi);
			event = (Event) oi.readObject();

		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (bi != null) {
					bi.close();
				}
				if (oi != null) {
					oi.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		return event;
	}
}
